package org.deng.learn.lambda;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.locks.LockSupport;

/**
 * @author 邓力宾
 * @date 2020/12/18 11:07
 * @desc 发布 订阅 使用jdk9
 */
public class FlowDemo {
    public static void main(String[] args) {
        //1.定义发布者，发布的数据类型是Integer
        //使用jdk自带的SubmissionPublisher,它实现了Publisher接口
        SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();

        //2.定义订阅者
        Flow.Subscriber<Integer> subscriber = new Flow.Subscriber<Integer>() {
            private Flow.Subscription subscription;
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                //保存订阅关系，需要用它来给发布者响应
                this.subscription = subscription;
                //请求一个数据
                this.subscription.request(1);
            }

            @Override
            public void onNext(Integer item) {
                //接收到一个数据
                System.out.println("接受到数据:"+item);

                //处理完后，再调用request请求一个数据
                this.subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                //出现异常 例如：处理数据时出现异常
                throwable.printStackTrace();
                //告诉发布者，后面不接收数据了
                this.subscription.cancel();
            }

            @Override
            public void onComplete() {
                //数据处理完成
                System.out.println("处理完成");
            }
        };

        //3.发布者和订阅者建立订阅关系
        publisher.subscribe(subscriber);

        //4.生产数据并发布
        int data = 11;
        publisher.submit(data);
        publisher.submit(data);
        publisher.submit(data);

        //5.结束后 关闭发布者
        //正式环境应该放到finally或者使用try-resource确保关闭
        publisher.close();
        //阻塞主线程，防止程序退出
        LockSupport.park();
    }
}
